package com.kiss.rpc.discovery.impl;

import com.kiss.rpc.Constant;
import com.kiss.rpc.RpcBootstrap;
import com.kiss.rpc.ServiceConfig;
import com.kiss.rpc.discovery.AbstractRegistry;
import com.kiss.rpc.exceptions.DiscoveryException;
import com.kiss.rpc.utils.NetUtils;
import com.kiss.rpc.utils.zookeeper.ZookeeperNode;
import com.kiss.rpc.utils.zookeeper.ZookeeperUtils;
import com.kiss.rpc.watcher.UpAndDownWatcher;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.List;

@Slf4j
public class ZooKeeperRegistry extends AbstractRegistry {

    private final ZooKeeper zk;

    private void initRooNode() {
        List<String> strings1 = Arrays.stream(Constant.BASE_PROVIDERS_PATH.split("/")).toList();
        List<String> strings2 = Arrays.stream(Constant.BASE_CONSUMERS_PATH.split("/")).toList();
        StringBuilder sb = new StringBuilder();
        for (List<String> strings : Arrays.asList(strings1, strings2)) {
            sb.delete(0, sb.length());
            for (String s : strings) {
                if (s.isEmpty()) continue;
                sb.append("/").append(s);
                if (ZookeeperUtils.notExists(zk, sb.toString(), null)) {
                    ZookeeperNode node = new ZookeeperNode(sb.toString(), null);
                    ZookeeperUtils.createNode(zk, node, null, CreateMode.PERSISTENT);
                }
            }
        }
    }

    public ZooKeeperRegistry() {
        this.zk = ZookeeperUtils.createZookeeper();
        initRooNode();
    }

    public ZooKeeperRegistry(String connectString, int timeout) {
        this.zk = ZookeeperUtils.createZookeeper(connectString, timeout);
        initRooNode();
    }

    @Override
    public void register(ServiceConfig<?> service) {
        String parentNode = Constant.BASE_PROVIDERS_PATH + "/" + service.getInterface().getName();
        // 服务节点
        if (ZookeeperUtils.notExists(zk, parentNode, null)) {
            ZookeeperNode node = new ZookeeperNode(parentNode, null);
            ZookeeperUtils.createNode(zk, node, null, CreateMode.PERSISTENT);
        }
        // 分组节点
        parentNode = parentNode + "/" + service.getGroup();
        if (ZookeeperUtils.notExists(zk, parentNode, null)) {
            ZookeeperNode node = new ZookeeperNode(parentNode, null);
            ZookeeperUtils.createNode(zk, node, null, CreateMode.PERSISTENT);
        }

        // 创建本机的临时节点
        String node = parentNode + "/" + NetUtils.getIp() + ":" + RpcBootstrap.getInstance().getConfig().getPort();
        if (ZookeeperUtils.notExists(zk, node, null)) {
            ZookeeperNode _node = new ZookeeperNode(node, null);
            ZookeeperUtils.createNode(zk, _node, null, CreateMode.PERSISTENT);
        }

        if(log.isDebugEnabled()){
            log.debug("服务{}，已经被注册", service.getInterface().getName());
        }
    }

    @Override
    public List<InetSocketAddress> lookup(String serviceName, String group) {
        // 1、找到服务对应的节点
        String serviceNode = Constant.BASE_PROVIDERS_PATH + "/" + serviceName + "/" +group;

        // 2、从zk中获取他的子节点, 192.168.12.123:2151
        List<String> children = ZookeeperUtils.getChildren(zk, serviceNode, new UpAndDownWatcher());
        // 获取了所有的可用的服务列表
        List<InetSocketAddress> inetSocketAddresses = children.stream().map(ipString -> {
            String[] ipAndPort = ipString.split(":");
            String ip = ipAndPort[0];
            int port = Integer.parseInt(ipAndPort[1]);
            return new InetSocketAddress(ip, port);
        }).toList();

        if(inetSocketAddresses.isEmpty()){
            throw new DiscoveryException("未发现任何可用的服务主机.");
        }

        return inetSocketAddresses;
    }

    @Override
    public void unregister(String serviceName, String group, String ipAndPort) {
        if (serviceName == null || ipAndPort == null) return;

        if (ipAndPort.startsWith("/")) ipAndPort = ipAndPort.substring(1);
        String nodePath = Constant.BASE_PROVIDERS_PATH + "/" + serviceName + "/" + group + "/" +  ipAndPort;
        ZookeeperUtils.deleteNode(zk, nodePath);
    }

}
